package com.xiaoshuidi.cloud.module.rooms.job.delayjob;

import com.alibaba.fastjson.JSONObject;
import com.xiaoshuidi.cloud.framework.tenant.core.context.TenantContextHolder;
import com.xiaoshuidi.cloud.module.rooms.enums.DelayJobType.DelayJobTypeEnum;
import com.xiaoshuidi.cloud.module.rooms.job.delayjob.strategy.IStrategy;
import com.xiaoshuidi.cloud.module.rooms.job.delayjob.strategy.ProcessingTask;
import com.xiaoshuidi.cloud.module.rooms.vo.DelayJobDTO;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author hans.zhang
 * @description 延迟job service
 * @date 2020/11/16
 */
@Component
@Slf4j
public class DelayJobService {

    private static final String delayJobName = "delay_job";

    @Autowired
    private RedissonClient redissonClient;

    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    @PostConstruct
//    @DependsOn(value = {"redissonClient"})
    public void startJobListener() {
        log.info("========================开始执行DelayJobService延迟job=========================================");
        ListenerTask task = new ListenerTask(redissonClient);
        executorService.submit(task);
        // 系统重启后,发起一次任务消费, 避免出现队列中有到期但是没有被放到目标队列的任务, 避免任务未处理
        addDelayJob(DelayJobTypeEnum.INIT,"", 1, TimeUnit.SECONDS);
    }

    /**
     * 添加延迟任务
     *
     * @param delayJob
     * @param delay    延迟时间
     * @param timeUnit 时间单位
     */
    public void addDelayJob(DelayJobTypeEnum delayJobTypeEnum, Object delayJob, long delay, TimeUnit timeUnit) {
        RBlockingQueue<Object> blockingFairQueue = redissonClient.getBlockingQueue(delayJobName);
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        DelayJobDTO delayJobDTO = new DelayJobDTO().setDelayJobTypeEnum(delayJobTypeEnum).setDelayJob(delayJob).setTenantId(TenantContextHolder.getTenantId());
        delayedQueue.offer(delayJobDTO, delay, timeUnit);
    }


    /**
     * 内部类，ListenerTask，执行具体的队列消费
     */
    class ListenerTask implements Runnable {

        private RedissonClient redissonClient;

        private ListenerTask(RedissonClient redissonClient) {
            this.redissonClient = redissonClient;
        }

        @Override
        public void run() {
            RBlockingQueue<DelayJobDTO> blockingFairQueue = redissonClient.getBlockingQueue(delayJobName);
            while (true) {
                try {
                    if(redissonClient.isShutdown()){
                        break;
                    }
                    DelayJobDTO delayJob = blockingFairQueue.take();
                    // 因为从redis队列中拿数据执行没有租户id, 因此手动插入租户id
                    TenantContextHolder.setTenantId(delayJob.getTenantId());
                    if (!DelayJobTypeEnum.INIT.equals(delayJob.getDelayJobTypeEnum())) {
                        ProcessingTask processingTask = new ProcessingTask();
                        IStrategy strategy = processingTask.getStrategy(delayJob.getDelayJobTypeEnum());
                        strategy.consumerTask(delayJob.getDelayJob());
                        log.info("delayJob执行结果:{}", delayJob);
                    }
                } catch (Exception e) {
                    log.warn(e.getMessage(), e);
                } finally {
                    TenantContextHolder.clear();
                }
            }
        }
    }
}
